#ifndef COROUTINE_CHANNEL_H
#define COROUTINE_CHANNEL_H

#include "../base/Defind.hpp"
#include "../base/BaseLog.hpp"
#include "Fiber.hpp"
#include <asm-generic/errno.h>
#include <cerrno>
#include <cstring>
#include <exception>
#include <list>
#include <mutex>
#include <queue>
#include <string>
#include <unistd.h>

namespace DBServer {
namespace coroutine
{

/// @brief 协程之间的同步机制
class Channel {
public:
    DISABLE_COPY(Channel);

    Channel();

    Channel(Channel &&);

    Channel &operator=(Channel &&);

    ~Channel();

    template<typename input>
    void operator<(input in) {
        std::unique_lock<std::mutex> lock(head_mtx);
        // 如果阻塞队列为空，则没有协程在这个channel上睡眠
        if (blocking_queue.empty()) {
            return;
        }

        try
        {
            std::string in_str = std::to_string(in);
            int rt = write(chan[1], in_str.c_str(), 1);
            if (rt < -1)
            {
                error("what():%s\n", strerror(errno));
                throw "";
            }

            // 唤醒一个协程
            auto hold_fiber = blocking_queue.front();
            hold_fiber->set_state(FiberState::READY);
            blocking_queue.pop();
        }
        catch (const std::exception& e)
        {
            error(e.what());
            close(chan[1]);
            chan[1] = -1;
            close(chan[0]);
            chan[0] = -1;
            throw;
        }
    }

    template<typename output>
    void operator>(output out) {
        try
        {
            // 将协程放入阻塞队列进行等待
            push();
            Fiber::Yield();

            int rt = read(chan[0], (void *)out, 1);
            if (rt < 0)
            {
                error("read bytes(%d) fd(%d) what(%d): %s\n", rt, chan[0], errno, strerror(errno));
                throw "";
            }
        }
        catch (const std::exception& e)
        {
            error(e.what());
            close(chan[1]);
            chan[1] = -1;
            close(chan[0]);
            chan[0] = -1;
            throw;
        }
    }
private:
    void push();

    void pop();

private:
    int chan[2];
    std::mutex head_mtx, tail_mtx;
    std::queue<Fiber::ptr, std::list<Fiber::ptr>> blocking_queue;
};

template <>
void Channel::operator< <char *>(char *in);

template<>
void Channel::operator< <std::string>(std::string in);

template <>
void Channel::operator< <std::string &>(std::string &in);

template <>
void Channel::operator< <std::string &&>(std::string &&in);


}   // coroutine   
}   // DBServer
#endif // !COROUTINE_CHANNEL_H